package com.yf.dxapp.module.app.config;

import cn.hutool.system.SystemUtil;
import com.yf.dxapp.module.app.redis.AbstractRedisStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.util.List;

@Configuration
@Slf4j
public class RedisConfig {


    /**
     * https://docs.spring.io/spring-data/redis/reference/redis/redis-streams.html
     * @param connectionFactory
     * @return
     */
    @Bean(initMethod = "start",destroyMethod = "stop")
    public StreamMessageListenerContainer streamMessageListenerContainer(RedisTemplate redisTemplate, RedisConnectionFactory connectionFactory, List<AbstractRedisStreamListener> listenerList){
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> containerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                .batchSize(10) //一次最多拉取十条
                .targetType(String.class) //统一String
                .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
                containerOptions);
        String consumerName = buildConsumerName();
        //parallelStream 并行多线程
        listenerList.parallelStream().forEach(listener->{
            listener.setRedisTemplate(redisTemplate);
            //创建组的消费方式 重启的时候 之前已经创建过组了 不用重复创建
            try {
                redisTemplate.opsForStream().createGroup(listener.getStreamKey(), listener.getGroup());
            }catch (Exception e){

            }
            //创建消费者
            Consumer consumer = Consumer.from(listener.getGroup(),consumerName);
            // 设置 Consumer 消费进度，以最小消费进度为准
            StreamOffset<String> streamOffset = StreamOffset.create(listener.getStreamKey(),ReadOffset.lastConsumed());
            StreamMessageListenerContainer.ConsumerStreamReadRequestBuilder builder = StreamMessageListenerContainer.StreamReadRequest.builder(streamOffset)
                    .consumer(consumer)
                    .autoAcknowledge(false)//不自动确认
                    .cancelOnError(throwable -> false);//发生异常不丢掉消息
            container.register(builder.build(),listener);
            log.info("[redisStreamMessageListenerContainer][完成注册 StreamKey({}) 对应的监听器({})]",
                    listener.getStreamKey(), listener.getClass().getName());
        });

        return container;

    }

    /**
     * 构建消费者名字，使用本地 IP + 进程编号的方式。
     * 参考自 RocketMQ clientId 的实现
     *
     * @return 消费者名字
     */
    private static String buildConsumerName() {
        return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID());
    }

}
